package com.gitee.jnxj.nraft.logreply;

import com.gitee.jnxj.nraft.common.log.LogEntry;
import com.gitee.jnxj.nraft.common.log.LogSerializeUtil;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * 日志管理器
 *
 * @author 江南小俊
 * @since 2021/7/13
 **/
@RequiredArgsConstructor
@Slf4j
public class LogManager {
    @Getter
    private static LogManager instance;
    private String path;
    private RocksDB rocksDB;

    public LogManager(String path) {
        this.path = path;
        instance = this;
    }

    @PostConstruct
    public void init() {
        log.debug("RocksDB init...");
        RocksDB.loadLibrary();
        try {
            rocksDB = RocksDB.open(path);
        } catch (RocksDBException e) {
            log.error("打开日志数据库：{},", path, e);
        }
    }

    @PreDestroy
    public void destroy() {
        log.debug("RocksDB destroy...");
        rocksDB.close();
    }

    /**
     * 添加日志条目：从1开始
     *
     * @param entry
     * @return boolean
     * @author Jim
     * @since 2021/7/13 下午4:35
     **/

    public synchronized boolean appendEntry(LogEntry entry) {
        //找出最新的日志条目
        RocksIterator rocksIterator = rocksDB.newIterator();
        rocksIterator.seekToLast();
        //已验证
        if (rocksIterator.isValid()) {
            byte[] key = rocksIterator.key();
            LogEntry.LogKey logKey = LogSerializeUtil.deserialize(key);
            logKey.setKeyId(logKey.getKeyId() + 1);
            entry.setKey(logKey);
        } else {
            entry.setKey(LogEntry.LogKey.builder().keyId(1).build());
        }
        log.debug("appendEntry:{}", entry);
        try {
            rocksDB.put(LogSerializeUtil.serialize(entry.getKey()), LogSerializeUtil.serialize(entry.getValue()));
        } catch (RocksDBException e) {
            log.error("", e);
            return false;
        }
        return true;
    }

    /**
     * 找出最新的日志条目(提交或未提交)
     *
     * @return com.gitee.jnxj.nraft.common.log.LogEntry
     * @author Jim
     * @since 2021/7/13 下午5:14
     **/

    public LogEntry getLastEntry() {
        //找出最新的日志条目
        RocksIterator rocksIterator = rocksDB.newIterator();
        rocksIterator.seekToLast();
        if (rocksIterator.isValid()) {
            return buildLogEntry(rocksIterator);
        } else {
            return null;
        }
    }

    /**
     * 找出最新已提交的日志条目
     *
     * @return com.gitee.jnxj.nraft.common.log.LogEntry
     * @author Jim
     * @since 2021/7/13 下午5:14
     **/

    public LogEntry getLastEntryOfCommit() {
        //找出最新的日志条目
        RocksIterator rocksIterator = rocksDB.newIterator();
        for (rocksIterator.seekToLast(); rocksIterator.isValid(); rocksIterator.prev()) {
            LogEntry.LogValue logValue = LogSerializeUtil.deserialize(rocksIterator.value());
            if (logValue.isCommit()) {
                return buildLogEntry(rocksIterator);
            }
        }
        return null;
    }

    private LogEntry buildLogEntry(RocksIterator rocksIterator) {
        return LogEntry.builder()
                .key(LogSerializeUtil.deserialize(rocksIterator.key()))
                .value(LogSerializeUtil.deserialize(rocksIterator.value()))
                .build();
    }

    /**
     * 提交日志
     *
     * @param keyId
     * @return boolean
     * @author Jim
     * @since 2021/7/13 下午5:14
     **/

    public boolean commit(long keyId) {
        RocksIterator rocksIterator = rocksDB.newIterator();
        for (rocksIterator.seekToLast(); rocksIterator.isValid(); rocksIterator.prev()) {
            LogEntry.LogKey logKey = LogSerializeUtil.deserialize(rocksIterator.key());
            if (logKey.getKeyId() > keyId) continue;
            LogEntry.LogValue logValue = LogSerializeUtil.deserialize(rocksIterator.value());
            if (logValue.isCommit()) {
                return true;
            } else {
                logValue.setCommit(true);
                try {
                    rocksDB.put(LogSerializeUtil.serialize(logKey), LogSerializeUtil.serialize(logValue));
                } catch (RocksDBException e) {
                    log.error("", e);
                    return false;
                }
            }
        }
        return true;
    }

    /**
     * 提交Entry
     *
     * @param entry
     * @return boolean
     * @author Jim
     * @since 2021/7/13 下午5:37
     **/

    public boolean commitWithEntry(LogEntry entry) {
        RocksIterator rocksIterator = rocksDB.newIterator();
        for (rocksIterator.seekToLast(); rocksIterator.isValid(); rocksIterator.prev()) {
            LogEntry.LogKey logKey = LogSerializeUtil.deserialize(rocksIterator.key());
            LogEntry.LogValue logValue = LogSerializeUtil.deserialize(rocksIterator.value());
            if (logKey.getKeyId() > entry.getKey().getKeyId()) {
                //如果没有提交->删除
                if (!logValue.isCommit()) {
                    try {
                        rocksDB.delete(LogSerializeUtil.serialize(logKey));
                    } catch (RocksDBException e) {
                        log.error("", e);
                    }
                } else {
                    return true;
                }
            } else if (logKey.getKeyId() == entry.getKey().getKeyId()) {
                if (!logValue.isCommit()) {
                    entry.getValue().setCommit(true);
                    try {
                        rocksDB.put(LogSerializeUtil.serialize(logKey), LogSerializeUtil.serialize(entry.getValue()));
                    } catch (RocksDBException e) {
                        log.error("", e);
                        return false;
                    }
                } else {
                    return true;
                }
            } else {
                if (!logValue.isCommit()) {
                    logValue.setCommit(true);
                    try {
                        rocksDB.put(LogSerializeUtil.serialize(logKey), LogSerializeUtil.serialize(logValue));
                    } catch (RocksDBException e) {
                        log.error("", e);
                        return false;
                    }
                } else {
                    return true;
                }
            }
        }
        return true;
    }

    public LogEntry getEntryByKey(LogEntry.LogKey logKey) {
        RocksIterator rocksIterator = rocksDB.newIterator();
        for (rocksIterator.seekToLast(); rocksIterator.isValid(); rocksIterator.prev()) {
            LogEntry.LogKey key = LogSerializeUtil.deserialize(rocksIterator.key());
            if (key.equals(logKey)) {
                return buildLogEntry(rocksIterator);
            }
        }
        return null;
    }

    /**
     * 现有条目与新条目（索引相同但任期不同）发生冲突，删除当前及以后的所有条目<br/>
     * 添加不在日志中的新条目
     *
     * @param newEntry
     * @return void
     * @author Jim
     * @since 2021/7/14 下午5:16
     **/

    public void addEntry(LogEntry newEntry) {
        boolean conflict = false;
        RocksIterator iterator = rocksDB.newIterator();
        for (iterator.seekToLast(); iterator.isValid(); iterator.prev()) {
            LogEntry logEntry = buildLogEntry(iterator);
            if (logEntry.getKey().getKeyId() == newEntry.getKey().getKeyId()) {
                if (logEntry.getValue().getTerm() == newEntry.getValue().getTerm()) {
                    //有日志，没有冲突，直接返回
                    return;
                } else {
                    //有冲突
                    conflict = true;
                    break;
                }
            }
            //没有日志，没有冲突
            if (logEntry.getKey().getKeyId() < newEntry.getKey().getKeyId()) {
                break;
            }

        }
        if (conflict) {
            //冲突
            for (; iterator.isValid(); iterator.next()) {
                try {
                    rocksDB.delete(LogSerializeUtil.serialize(iterator.key()));
                } catch (RocksDBException e) {
                    log.error("", e);
                }
            }
        }
        //添加
        try {
            rocksDB.put(LogSerializeUtil.serialize(newEntry.getKey()), LogSerializeUtil.serialize(newEntry.getValue()));
        } catch (RocksDBException e) {
            log.error("", e);
        }
    }
}
